package com.easou.ps.master.search;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import org.apache.log4j.Logger;

import com.easou.ps.common.servicehandle.ServiceAddress;
import com.easou.ps.master.rpc.KSRPCProxy;
import com.easou.ps.master.rpc.RPCProxyFactory;
import com.easou.ps.master.rpc.ServerAddressFactory;
import com.easou.ps.thrift.ks.KnowledgeSearchRequest;
import com.easou.ps.thrift.ks.KnowledgeSearchResponse;

public class KSTask implements Callable<KnowledgeSearchResponse>{
	
	static final Logger LOG = Logger.getLogger("search");
	private KnowledgeSearchRequest query;
	private SearchTarget target;
	private Map<String,String> workerIp;
	private long timeCost = 0;
	
	public long getTimeCost() {
		return timeCost;
	}

	public void setTimeCost(long timeCost) {
		this.timeCost = timeCost;
	}

	public KSTask(KnowledgeSearchRequest query, SearchTarget target, Map<String,String> workerIp) {
		this.query = query;
		this.target = target;
		this.workerIp = workerIp;
	}
	
	@Override
	public KnowledgeSearchResponse call() throws Exception {
		long startTime = System.currentTimeMillis();
		List<ServiceAddress> addresses = ServerAddressFactory.selectAddresses(target.getType().toString());
		LOG.info("[select worker] type:" + target.getType().toString() + ", addresses:" + addresses);
		if (addresses == null || addresses.size() == 0) {
			if (workerIp.get(target.getType().toString())==null) {				
				workerIp.put(target.getType().toString(), "null");
			}
			return new KnowledgeSearchResponse();
		}
		KnowledgeSearchResponse response = null;
		if (addresses.size() > 0) {
			ServiceAddress addr = addresses.get(0);
			if (workerIp.get(target.getType().toString())==null) {				
				workerIp.put(target.getType().toString(), addr.getHost()+":"+addr.getPort());
			}
			final KSRPCProxy proxy = RPCProxyFactory.getRPCKsProxy(addr);
			response = proxy.search(query);
		}
		setTimeCost(System.currentTimeMillis() - startTime);
		return response;
	}
}
